home
***
CD-ROM
|
disk
|
FTP
|
other
***
search
/
Cream of the Crop 26
/
Cream of the Crop 26.iso
/
os2
/
pvm34b3.zip
/
pvm34b3
/
pvm3
/
src
/
tdpro.c
< prev
next >
Wrap
C/C++ Source or Header
|
1997-08-08
|
56KB
|
2,753 lines
static char rcsid[] =
"$Id: tdpro.c,v 1.20 1997/07/02 20:27:35 pvmsrc Exp $";
/*
* PVM version 3.4: Parallel Virtual Machine System
* University of Tennessee, Knoxville TN.
* Oak Ridge National Laboratory, Oak Ridge TN.
* Emory University, Atlanta GA.
* Authors: J. J. Dongarra, G. E. Fagg, M. Fischer
* G. A. Geist, J. A. Kohl, R. J. Manchek, P. Mucci,
* P. M. Papadopoulos, S. L. Scott, and V. S. Sunderam
* (C) 1997 All Rights Reserved
*
* NOTICE
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby granted
* provided that the above copyright notice appear in all copies and
* that both the copyright notice and this permission notice appear in
* supporting documentation.
*
* Neither the Institutions (Emory University, Oak Ridge National
* Laboratory, and University of Tennessee) nor the Authors make any
* representations about the suitability of this software for any
* purpose. This software is provided ``as is'' without express or
* implied warranty.
*
* PVM version 3 was funded in part by the U.S. Department of Energy,
* the National Science Foundation and the State of Tennessee.
*/
/*
* tdpro.c
*
* Entry points for messages from local tasks.
*
$Log: tdpro.c,v $
* Revision 1.20 1997/07/02 20:27:35 pvmsrc
* Fixed startup race on shmem to that a shmem task can get fully
* configured before getting any messages.
* This involved adding two states
* TF_PRESHMCONN and TF_SHM. TF_PRESHMCONN indicates that messages
* with MM_PRIO set can be sent to a task, but regular messages are
* queued. This allows shmem tasks to be completely configured
* before any messages flow. When the daemon changes the state from
* TF_PRESHMCONN to TF_SHMCONN it calls shm_wrt_pkts to write any
* packets that were queued before task state changed to TF_SHMCONN.
*
* Revision 1.19 1997/06/27 17:32:56 pvmsrc
* Updated for WIN32 header files & Authors.
*
* Revision 1.18 1997/06/27 16:10:01 pvmsrc
* Spawning tasks with option PvmNoSpawnParent get the
* parent tid set to PvmParentNotSet in spawned tasks.
*
* Revision 1.17 1997/06/24 20:39:22 pvmsrc
* Eliminated unnecessary global externs.
*
* Revision 1.16 1997/06/23 21:26:46 pvmsrc
* Added checking for master host in tm_hoster().
* - actually check hd_hostpart instead of ht_local/ht_master indices,
* in case freaking host table has copies/duplicates.
* (always wear 2 condoms... :-)
* - return new PvmHostrNMstr error code.
*
* Revision 1.15 1997/06/16 13:41:44 pvmsrc
* assign_tasks() now passes distribution info in DM_EXEC message.
*
* Revision 1.14 1997/06/02 13:20:48 pvmsrc
* tm_shmconn() marks old task w/ TF_CLOSE
* - so work() doesn't complain about a dead socket.
*
* Revision 1.13 1997/05/28 14:04:05 pvmsrc
* Cleaned up TM_SHMCONN() handler routine.
* - formatted like hell, and didn't compile clean on SGI64.
*
* Revision 1.12 1997/05/27 14:45:53 pvmsrc
* Added new TM message TM_SHMCONN with routine tm_shmconn():
* which merges records and sets TF_SHMCONN flag.
* Updated entry point in loclentry() as well:
* to accept this type of connection.
*
* Revision 1.11 1997/05/12 20:28:22 pvmsrc
* Removed duplicate #includes...
*
* Revision 1.10 1997/05/07 21:22:12 pvmsrc
* AAAIIIEEEEEEEEE!!!!
* - removed all static-limited string unpacking:
* * replaced with use of:
* upkstralloc() / PVM_FREE() (for pvmd stuff).
* new pvmupkstralloc() / PVM_FREE() (for lpvm stuff).
* * manual allocation of local buffers for sprintf() & packing.
* * alternate static string arrays to replace fixed-length cases.
* * I hope this shit works... :-Q
*
* Revision 1.9 1997/04/30 21:26:31 pvmsrc
* SGI Compiler Warning Cleanup.
*
* Revision 1.8 1997/03/27 19:57:10 pvmsrc
* Spanked in PvmNoSpawnParent flag usage.
* - in tm_spawn(), don't set wxp->w_ptid if PvmNoSpawnParent set.
* - in assign_tasks(), use wxp->w_ptid as parent tid instead of
* snarfing it from wp->wa_tid... (Bob Hack) D-Oh...
*
* Revision 1.7 1997/02/13 15:10:05 pvmsrc
* Removed unnecessary extern for struct waitc *waitlist.
* - now in global.h.
*
* Revision 1.6 1997/01/28 19:27:49 pvmsrc
* New Copyright Notice & Authors.
*
* Revision 1.5 1996/12/30 13:37:49 pvmsrc
* Fixed closing comment in last log entry...
*
* Revision 1.4 1996/12/20 15:02:07 pvmsrc
* Fixed missing comment end *-/ ... D-Oh!
*
* Revision 1.3 1996/10/25 13:58:08 pvmsrc
* Replaced old #includes for protocol headers:
* - <pvmsdpro.h>, "ddpro.h", "tdpro.h"
* With #include of new combined header:
* - <pvmproto.h>
*
* Revision 1.2 1996/10/24 22:11:04 pvmsrc
* Moved #include "global.h" below other #include's for typing.
* Added #include <pvmtev.h> for tracing constants.
* Added extern struct Pvmtracer pvmtracer for tracer info.
* Added new tm_tracer() routine to handle tracer registration requests
* from pvm_reg_tracer().
* In tm_conn2(), insert tracer info for tasks spawned from the shell.
* - can't get env vars set here, so rely on mailbox for that info.
* - use new tev_send_newtask() routine.
* In assign_tasks():
* - replace inline trace event code with new tev_send_spntask().
* - check trctid / outtid with > 0, not non-zero, for new case where
* task denies external collection.
* Modified change_trace() to use new event routines:
* - tev_send_newtask(), tev_send_spntask() & tev_send_endtask().
*
* Revision 1.1 1996/09/23 23:44:45 pvmsrc
* Initial revision
*
* Revision 1.9 1995/07/19 21:27:26 manchek
* use new function pvmnametag instead of [dts]mname
*
* Revision 1.8 1995/07/03 19:11:40 manchek
* added tickle #100
*
* Revision 1.7 1995/05/17 16:55:20 manchek
* changed global mytid to pvmmytid.
* added TM_GETOPT message class
*
* Revision 1.6 1994/10/15 19:37:46 manchek
* cast message tags for comparison as integers.
* check newhosts when deleting host.
* unset TF_FORKD if task reconnects with a different pid
*
* Revision 1.5 1994/06/21 18:30:00 manchek
* subscript arith in tmname() broke with opt
*
* Revision 1.4 1994/06/04 21:45:54 manchek
* misc
*
* Revision 1.3 1994/06/03 20:38:31 manchek
* version 3.3.0
*
* Revision 1.2 1993/11/30 15:56:23 manchek
* tm_conn() complains if it can't write t-auth file (fs full?)
*
* Revision 1.1 1993/08/30 23:26:52 manchek
* Initial revision
*
*/
#ifdef NEEDMENDIAN
#include <machine/endian.h>
#endif
#ifdef NEEDENDIAN
#include <endian.h>
#endif
#ifdef NEEDSENDIAN
#include <sys/endian.h>
#endif
#ifndef WIN32
#include <rpc/types.h>
#include <rpc/xdr.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#else
#include "..\xdr\types.h"
#include "..\xdr\xdr.h"
#include <sys/types.h>
#include <time.h>
#endif
#include <fcntl.h>
#ifdef SYSVSTR
#include <string.h>
#else
#include <strings.h>
#endif
#include <errno.h>
#include <stdio.h>
#include <pvm3.h>
#include <pvmproto.h>
#include "pvmalloc.h"
#include "pmsg.h"
#include "task.h"
#include "host.h"
#include "waitc.h"
#include "listmac.h"
#include "bfunc.h"
#include <pvmtev.h>
#include "global.h"
#ifndef min
#define min(a,b) ((a)<(b)?(a):(b))
#endif
#ifdef NOTMPNAM
#define TMPNAMFUN(x) pvmtmpnam(x)
#define LEN_OF_TMP_NAM 32
char *pvmtmpnam();
#else /*NOTMPNAM*/
#define TMPNAMFUN(x) tmpnam(x)
#ifdef L_tmpnam
#define LEN_OF_TMP_NAM L_tmpnam
#else
#define LEN_OF_TMP_NAM 64
#endif
#endif /*NOTMPNAM*/
/***************
** Globals **
** **
***************/
extern void task_dump();
char *debug_flags();
char *inadport_hex();
void i_dump();
void tev_send_newtask();
void tev_send_spntask();
void tev_send_endtask();
extern int pvmdebmask; /* from pvmd.c */
extern int hostertid; /* from pvmd.c */
extern struct htab *hosts; /* from pvmd.c */
extern int pvmmydsig; /* from pvmd.c */
extern int myhostpart; /* from pvmd.c */
extern int pvmmytid; /* from pvmd.c */
extern struct htab *newhosts; /* from pvmd.c */
extern int nopax; /* from pvmd.c */
extern int pvmudpmtu; /* from pvmd.c */
extern int pvmschedtid; /* from pvmd.c */
extern int taskertid; /* from pvmd.c */
extern struct Pvmtracer pvmtracer; /* from pvmd.c */
extern int tidhmask; /* from pvmd.c */
/***************
** Private **
** **
***************/
int tm_addhost();
int tm_config();
int tm_conn2();
int tm_connect();
int tm_context();
int tm_db();
int tm_delhost();
int tm_exit();
int tm_getopt();
int tm_halt();
int tm_hoster();
int tm_hostsync();
int tm_mca();
int tm_mstat();
int tm_notify();
int tm_pstat();
int tm_sendsig();
int tm_sched();
int tm_setopt();
int tm_spawn();
int tm_task();
int tm_tasker();
int tm_tickle();
int tm_tracer();
int tm_shmconn();
int (*loclswitch[])() = {
tm_connect,
tm_conn2,
tm_exit,
tm_addhost,
tm_delhost,
tm_config,
tm_mstat,
tm_halt,
tm_tickle,
tm_spawn,
tm_pstat,
tm_sendsig,
tm_task,
tm_mca,
tm_notify,
tm_db,
tm_sched,
tm_tasker,
tm_hoster,
tm_tracer,
tm_hostsync,
tm_setopt,
tm_getopt,
tm_context,
tm_shmconn,
};
int
loclentry(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int c = mp->m_tag;
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf("loclentry() from t%x tag %s\n", mp->m_src,
pvmnametag(c, (int *)0));
/*
pvmhdump(mp->m_cfrag->fr_dat, mp->m_cfrag->fr_len, "frag: ");
*/
}
/*
if (mp->m_enc != 1) {
pvmlogprintf("loclentry() message from t%x with bad enc %d\n",
tp->t_tid, mp->m_enc);
goto bail;
}
*/
if (c < (int)TM_FIRST || c > (int)TM_LAST) {
pvmlogprintf("loclentry() message from t%x with bogus tag %d\n",
tp->t_tid, c);
goto bail;
}
if ((!(tp->t_flag & TF_CONN) && c != TM_CONNECT && c != TM_CONN2
&& c != TM_SHMCONN)
|| ((tp->t_flag & TF_AUTH) && c != TM_CONN2)) {
pvmlogerror("loclentry() non-connect message from anon task\n");
tp->t_flag |= TF_CLOSE;
goto bail;
}
c -= TM_FIRST;
(loclswitch[c])(tp, mp);
bail:
pmsg_unref(mp);
return 0;
}
/* replymessage()
*
* Construct reply message to sender with matching context, tag, wait id
*/
struct pmsg *
replymessage(mp)
struct pmsg *mp;
{
struct pmsg *rmp;
if (rmp = mesg_new(0)) {
rmp->m_dst = mp->m_src;
rmp->m_ctx = mp->m_ctx;
rmp->m_tag = mp->m_tag;
rmp->m_wid = mp->m_wid;
}
return rmp;
}
/*********************************
** Task request entry points **
** **
*********************************/
/* tm_connect()
*
* Task connecting to pvmd phase 1.
* We assign it a context, write in t-auth file to prove our ident,
* then make d-auth file for task to prove itself.
*
* TM_CONNECT
* call {
* int tdprotocol // t-d protocol number compiled into libpvm
* string authfile // t-auth file pvmd must write
* }
* ret {
* int tdprotocol // pvmd's td protocol version
* int acknack // 1 if pvmd accepts connection
* string authfile // d-auth file task must write
* }
*/
int
tm_connect(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int ver; /* task's libpvm t-d proto version */
char authfn[LEN_OF_TMP_NAM]; /* t-auth file name */
int d;
int cc;
if (upkint(mp, &ver) || upkstr(mp, authfn, sizeof(authfn))) {
pvmlogerror("tm_connect() bad msg format\n");
goto bail;
}
/*
* if protocols are not compatible, send nack
* context will get flushed after reply is sent
*/
if (ver != TDPROTOCOL) {
pvmlogprintf("tm_connect() t-d protocol mismatch (%d/%d)\n",
ver, TDPROTOCOL);
mp = replymessage(mp);
pkint(mp, TDPROTOCOL);
pkint(mp, 0);
pkstr(mp, "");
mp->m_flag |= MM_PRIO;
mesg_to_task(tp, mp);
pmsg_unref(mp);
goto bail;
}
/*
* write in t-auth file, create empty d-auth file that task
* must write and we'll check later
*/
if ((d = open(authfn, O_WRONLY, 0)) == -1) {
pvmlogperror("tm_connect() can't open t-auth file");
goto bail;
}
cc = write(d, authfn, 1);
if (cc != 1) {
if (cc == -1)
pvmlogperror(authfn);
pvmlogerror("tm_connect() can't write t-auth file\n");
}
(void)close(d);
tp->t_authnam = TALLOC(LEN_OF_TMP_NAM, char, "auth");
(void)TMPNAMFUN(tp->t_authnam);
#ifndef IMA_OS2
if ((tp->t_authfd = open(tp->t_authnam, O_RDONLY|O_CREAT|O_TRUNC, 0600))
#else
if ((tp->t_authfd = open(tp->t_authnam, O_RDWR|O_CREAT|O_TRUNC, 0600))
#endif
== -1) {
pvmlogperror("tm_connect() can't create d-auth file");
PVM_FREE(tp->t_authnam);
tp->t_authnam = 0;
goto bail;
}
/*
* task's turn to auth
*/
tp->t_flag |= TF_AUTH;
mp = replymessage(mp);
pkint(mp, TDPROTOCOL);
pkint(mp, 1);
pkstr(mp, tp->t_authnam);
mp->m_flag |= MM_PRIO;
mesg_to_task(tp, mp);
pmsg_unref(mp);
return 0;
bail:
tp->t_flag |= TF_CLOSE;
return 0;
}
/* tm_conn2()
*
* Task connecting to pvmd phase 2.
* We check d-auth file and give it config info, attach to real context.
*
* TM_CONN2
* call {
* int unixpid // real pid
* int cookie // magic cookie inherited from pvmd (or 0)
* }
* ret {
* int acknack // 1 if pvmd accepts connection
* int tid // task tid
* int ptid // parent tid
* int outtid // output dst
* int outctx
* int outtag
* int trctid // trace dst
* int trcctx
* int trctag
* int udpmax
* int nativecode
* string inaddr
* int schedtid // scheduler tid
* }
*/
int
tm_conn2(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int tid;
int pid; /* real pid of task */
int cookie; /* cookie to identify task or 0 */
struct task *tp2; /* to search for existing context */
struct pmsg *mp2;
int cc;
char c;
if (upkint(mp, &pid) || upkint(mp, &cookie)) {
pvmlogerror("tm_conn2() bad msg format\n");
goto bail;
}
if (!cookie)
cookie = pid;
if (!(tp->t_flag & TF_AUTH)) {
pvmlogprintf("tm_conn2() message from t%x, TF_AUTH not set\n",
tp->t_tid);
return 0;
}
/*
* check that task could write d-auth file
*/
if ((cc = read(tp->t_authfd, &c, 1)) == -1) {
pvmlogperror("tm_conn2() can't read d-auth file");
return 0;
}
if (cc != 1) {
pvmlogerror("tm_conn2() task didn't validate itself\n");
goto bail;
}
(void)close(tp->t_authfd);
tp->t_authfd = -1;
(void)unlink(tp->t_authnam);
PVM_FREE(tp->t_authnam);
tp->t_authnam = 0;
/*
* if task spawned by us, already has a context,
* else make it one
*/
if ((tp2 = task_findpid(cookie)) && !(tp2->t_flag & (TF_AUTH|TF_CONN))) {
if (pvmdebmask & PDMTASK) {
pvmlogprintf("tm_conn2() reconnect task t%x\n", tp2->t_tid);
}
tp->t_sched = tp2->t_sched;
} else {
if ((tid = tid_new()) < 0) {
pvmlogerror("tm_conn2() out of tids?\n");
goto bail; /* XXX should disconnect nicely */
}
tp2 = task_new(tid);
if (pvmschedtid) {
/* inform the scheduler about the new task */
tp->t_sched = pvmschedtid;
mp2 = mesg_new(0);
pkint(mp2, 1);
pkint(mp2, tid);
mp2->m_tag = SM_EXECACK; /* XXX yecch, overload */
mp2->m_dst = pvmschedtid;
sendmessage(mp2);
}
/* check for tracer */
if (pvmtracer.trctid && pvmtracer.trctag) {
tp2->t_trctid = pvmtracer.trctid;
tp2->t_trcctx = pvmtracer.trcctx;
tp2->t_trctag = pvmtracer.trctag;
/* Note: can't get the trace mask or trace buffer size
to task here, but this will be handled using mailbox */
/* trace start message */
/* XXX Note: we don't really know if we should send this
because this task could have prohibited tracing
by setting PvmSelfTraceTid to -1... */
tev_send_newtask(
pvmtracer.trctid, pvmtracer.trcctx, pvmtracer.trctag,
tid, -1, -1, "-" );
}
if (pvmtracer.outtid && pvmtracer.outtag) {
tp2->t_outtid = pvmtracer.outtid;
tp2->t_outctx = pvmtracer.outctx;
tp2->t_outtag = pvmtracer.outtag;
/* output start message */
/* Note: same deal here as for TEV_NEWTASK above... */
mp2 = mesg_new(0);
mp2->m_dst = pvmtracer.outtid;
mp2->m_ctx = pvmtracer.outctx;
mp2->m_tag = pvmtracer.outtag;
pkint(mp2, tid);
pkint(mp2, -2);
pkint(mp2, -1);
sendmessage(mp2);
}
if (pvmdebmask & PDMTASK) {
pvmlogprintf("tm_conn2() new task t%x\n", tp2->t_tid);
}
}
/*
* brundle-fly the contexts together
*/
tp2->t_sock = tp->t_sock;
tp2->t_sad = tp->t_sad;
tp2->t_salen = tp->t_salen;
if (tp2->t_pid != pid)
task_setpid(tp2, pid);
tp2->t_rxp = tp->t_rxp;
tp2->t_sched = tp->t_sched;
tp->t_sock = -1; /* tp will be freed by loclinput() */
tp->t_rxp = 0;
tp = tp2;
if (cookie != pid)
tp->t_flag &= ~TF_FORKD;
/*
* kick it in the butt; it's ready to go
*/
tp->t_flag &= ~TF_AUTH;
tp->t_flag |= TF_CONN;
mp = replymessage(mp);
pkint(mp, 1);
pkint(mp, tp->t_tid);
pkint(mp, tp->t_ptid);
pkint(mp, tp->t_outtid);
pkint(mp, tp->t_outctx);
pkint(mp, tp->t_outtag);
pkint(mp, tp->t_trctid);
pkint(mp, tp->t_trcctx);
pkint(mp, tp->t_trctag);
pkint(mp, pvmudpmtu);
pkint(mp, pvmmydsig);
pkstr(mp, inadport_hex(&(hosts->ht_hosts[hosts->ht_local]->hd_sad)));
pkint(mp, pvmschedtid);
mp->m_dst = tp->t_tid;
mp->m_flag |= MM_PRIO;
sendmessage(mp);
return 0;
bail:
tp->t_flag |= TF_CLOSE;
return 0;
}
/* tm_exit()
*
* Last message from a task. This is the nice way of disconnecting.
*
* TM_EXIT
* call { }
* ret { }
*/
int
tm_exit(tp, mp)
struct task *tp;
struct pmsg *mp;
{
mp = replymessage(mp);
tp->t_flag |= TF_CLOSE;
sendmessage(mp);
if (!(tp->t_flag & TF_FORKD))
task_cleanup(tp);
return 0;
}
/* tm_pstat()
*
* Task wants status of another task.
*
* TM_PSTAT
* call {
* int tid
* }
* ret {
* int status
* }
*/
int
tm_pstat(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int tid;
struct hostd *hp;
struct waitc *wp;
/* unpack and sanity check tid */
if (upkuint(mp, &tid)) {
pvmlogerror("tm_pstat() bad msg format\n");
return 0;
}
if (!TIDISTASK(tid)) {
pvmlogprintf("tm_pstat() bad tid %x\n", tid);
return 0;
}
/* nack if no such host */
if (!(hp = tidtohost(hosts, tid))) {
mp = replymessage(mp);
pkint(mp, PvmNoTask);
sendmessage(mp);
return 0;
}
/* else make a wait context and send query */
wp = wait_new(WT_PSTAT);
wp->wa_tid = tp->t_tid;
wp->wa_on = hp->hd_hostpart;
wp->wa_mesg = replymessage(mp);
mp = mesg_new(0);
mp->m_dst = hp->hd_hostpart | TIDPVMD;
mp->m_tag = DM_PSTAT;
mp->m_wid = wp->wa_wid;
pkint(mp, tid);
sendmessage(mp);
return 0;
}
/* tm_addhost()
*
* Task requesting to add hosts to virtual machine. Exit point is
* here or dm_addack().
*
* TM_ADDHOST
* call {
* int nhosts
* string names[nhosts]
* }
* ret {
* int nhosts // or error code
* int narches // if nhosts >= 0
* {
* int tid // or error code
* string name
* string arch
* int speed
* int dsig
* } [nhosts]
* }
*/
int
tm_addhost(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int count;
char *buf;
struct waitc *wp;
/* sanity check the message */
if (upkint(mp, &count))
goto bad;
if (count < 1 || count > (tidhmask >> (ffs(tidhmask) - 1)))
goto bad;
while (count-- > 0) {
if (upkstralloc(mp, &buf))
goto bad;
else
PVM_FREE(buf);
}
/* make a wait channel for the task */
wp = wait_new(WT_ADDHOST);
wp->wa_tid = tp->t_tid;
wp->wa_on = hosts->ht_hosts[hosts->ht_master]->hd_hostpart;
wp->wa_mesg = replymessage(mp);
/* forward message to master pvmd */
mp->m_ref++;
mp->m_src = pvmmytid;
mp->m_dst = hosts->ht_hosts[hosts->ht_master]->hd_hostpart | TIDPVMD;
mp->m_tag = DM_ADD;
mp->m_wid = wp->wa_wid;
sendmessage(mp);
return 0;
bad:
pvmlogprintf("tm_addhost() from t%x bad msg format\n", mp->m_src);
return 0;
}
int
free_wait_spawn(wxp)
struct waitc_spawn *wxp;
{
int i;
if (wxp->w_file)
PVM_FREE(wxp->w_file);
if (wxp->w_argv) {
for (i = 0; i < wxp->w_argc; i++)
if (wxp->w_argv[i])
PVM_FREE(wxp->w_argv[i]);
PVM_FREE(wxp->w_argv);
}
if (wxp->w_env) {
for (i = 0; i < wxp->w_nenv; i++)
if (wxp->w_env[i])
PVM_FREE(wxp->w_env[i]);
PVM_FREE(wxp->w_env);
}
if (wxp->w_ht)
ht_free(wxp->w_ht);
if (wxp->w_vec)
PVM_FREE(wxp->w_vec);
PVM_FREE(wxp);
return 0;
}
/* tm_spawn()
*
* Task requesting to spawn other tasks. Exit point for this
* request is here or dm_execack().
*
* TM_SPAWN
* call {
* string file
* int flags
* string where
* int count
* int nargs
* string argv[nargs]
* int outtid
* int outctx
* int outtag
* int trctid
* int trcctx
* int trctag
* int nenv
* string env[nenv]
* }
* ret {
* int ntids // or nack
* int tids[acknack]
* }
*/
int
tm_spawn(tp, mp)
struct task *tp;
struct pmsg *mp;
{
char *where = 0; /* location from req */
struct waitc *wp; /* 'seed' waitc */
struct waitc_spawn *wxp = 0; /* new task parameters */
struct htab *htp; /* set of usable hosts */
struct hostd *hp;
int hh;
int i;
/*
* unpack spawn command from task
*/
wxp = TALLOC(1, struct waitc_spawn, "waix");
BZERO((char*)wxp, sizeof(struct waitc_spawn));
if (upkstralloc(mp, &wxp->w_file)
|| upkint(mp, &wxp->w_flags)
|| upkstralloc(mp, &where)
|| upkint(mp, &wxp->w_veclen)
|| upkint(mp, &wxp->w_argc))
goto bad;
if (wxp->w_veclen < 1)
goto bad;
wxp->w_argv = TALLOC(wxp->w_argc + 1, char*, "argv");
BZERO((char*)wxp->w_argv, (wxp->w_argc + 1) * sizeof(char*));
for (i = 0; i < wxp->w_argc; i++)
if (upkstralloc(mp, &wxp->w_argv[i]))
goto bad;
if (upkuint(mp, &wxp->w_outtid)
|| upkuint(mp, &wxp->w_outctx)
|| upkuint(mp, &wxp->w_outtag)
|| upkuint(mp, &wxp->w_trctid)
|| upkuint(mp, &wxp->w_trcctx)
|| upkuint(mp, &wxp->w_trctag))
goto bad;
if (upkuint(mp, &wxp->w_nenv))
goto bad;
wxp->w_env = TALLOC((wxp->w_nenv + 1), char*, "env");
BZERO((char*)wxp->w_env, (wxp->w_nenv + 1) * sizeof(char*));
for (i = 0; i < wxp->w_nenv; i++)
if (upkstralloc(mp, &wxp->w_env[i]))
goto bad;
/*
* make host set containing hosts (matching where option)
*/
if ((wxp->w_flags & (PvmTaskHost|PvmTaskArch)) && !where)
goto bad;
htp = ht_new(1);
if (wxp->w_flags & PvmTaskHost) { /* given host */
if (hp = nametohost(hosts, where))
ht_insert(htp, hp);
} else {
if (wxp->w_flags & PvmTaskArch) { /* given arch */
for (hh = hosts->ht_last; hh > 0; hh--)
if ((hp = hosts->ht_hosts[hh])
&& !strcmp(where, hp->hd_arch))
ht_insert(htp, hp);
} else { /* anywhere */
ht_merge(htp, hosts);
}
}
if (wxp->w_flags & PvmHostCompl) {
for (hh = hosts->ht_last; hh > 0; hh--) {
if (hh <= htp->ht_last && (hp = htp->ht_hosts[hh]))
ht_delete(htp, hp);
else
if (hp = hosts->ht_hosts[hh])
ht_insert(htp, hp);
}
}
if (pvmdebmask & PDMTASK) {
pvmlogerror("tm_spawn() host set:\n");
ht_dump(htp);
}
if ( !(wxp->w_flags & PvmNoSpawnParent) ) {
wxp->w_ptid = tp->t_tid;
}
else
wxp->w_ptid = PvmParentNotSet; /* indicate that task unset parent */
/*
* assign each task to a host
*/
wxp->w_ht = htp;
wxp->w_vec = TALLOC(wxp->w_veclen, int, "vec");
BZERO((char*)wxp->w_vec, wxp->w_veclen * sizeof(int));
wxp->w_togo = wxp->w_veclen;
wp = wait_new(WT_SPAWN);
wp->wa_tid = tp->t_tid;
wp->wa_spec = (void*)wxp;
wxp = 0;
wp->wa_mesg = replymessage(mp);
assign_tasks(wp);
/* if already done, reply to task */
if (wp->wa_peer == wp) {
assign_tasks(wp);
}
wait_delete(wp);
goto cleanup;
bad:
pvmlogprintf("tm_spawn() from t%x bad msg format\n", mp->m_src);
cleanup:
if (where)
PVM_FREE(where);
if (wxp)
free_wait_spawn(wxp);
return 0;
}
/* assign_tasks()
*
* This is only called when no replies are pending (at the beginning
* of the operation or when all pvmds have checked back in).
*/
assign_tasks(wp)
struct waitc *wp; /* (any) waitc in peer group for this op */
{
static int lasthh = -1; /* for assigning hosts */
struct waitc_spawn *wxp = (struct waitc_spawn*)wp->wa_spec;
struct htab *htp; /* set of hosts to use */
int *vec; /* result/status vector */
int veclen; /* length of vector */
int count = 0; /* num of tasks to be assigned */
int nh; /* num of hosts to assign tasks */
int a = 0; /* accum for finding hosts */
int na = 0; /* num tasks assigned to a host */
struct waitc *wp2;
struct hostd *hp;
struct pmsg *mp;
int t;
int i;
int tid;
struct timeval now;
if (!wxp)
return 0;
htp = wxp->w_ht;
vec = wxp->w_vec;
veclen = wxp->w_veclen;
/*
* if no hosts left, fill unassigned entries with PvmNoHost
*/
if (!htp->ht_cnt)
for (t = veclen; t-- > 0; )
if (!vec[t])
vec[t] = PvmNoHost;
/*
* count tasks to be assigned, if none left reply to task
*/
for (t = veclen; t-- > 0; )
if (!vec[t])
count++;
if (!count) {
pkint(wp->wa_mesg, wxp->w_veclen);
for (t = 0; t < wxp->w_veclen; t++) {
tid = wxp->w_vec[t];
pkint(wp->wa_mesg, tid);
if (TIDISTASK(tid) && wxp->w_trctid > 0) {
tev_send_spntask(
wxp->w_trctid, wxp->w_trcctx, wxp->w_trctag,
tid, wxp->w_ptid );
}
if (TIDISTASK(tid) && wxp->w_outtid > 0) {
mp = mesg_new(0);
mp->m_dst = wxp->w_outtid;
mp->m_ctx = wxp->w_outctx;
mp->m_tag = wxp->w_outtag;
pkint(mp, tid);
pkint(mp, -1);
pkint(mp, wxp->w_ptid);
sendmessage(mp);
}
}
sendmessage(wp->wa_mesg);
wp->wa_mesg = 0;
free_wait_spawn(wxp);
wp->wa_spec = 0;
return 0;
}
/*
* assign tasks to hosts
*/
nh = min(htp->ht_cnt, count);
/* find first host to assign */
if (lasthh == -1)
lasthh = hosts->ht_local + 1;
if (lasthh > htp->ht_last)
lasthh = 0;
while (!htp->ht_hosts[lasthh])
if (++lasthh > htp->ht_last)
lasthh = 1;
hp = htp->ht_hosts[lasthh];
for (t = 0; t < veclen && vec[t]; t++);
while (t < veclen) {
/*
pvmlogprintf("assign_tasks() %s <- %d\n", hp->hd_name, t);
*/
vec[t] = hp->hd_hostpart;
na++;
/* when enough tasks for this host, move to next */
if ((a += nh) >= count) {
a -= count;
wp2 = wait_new(WT_SPAWN);
wp2->wa_tid = wp->wa_tid;
wp2->wa_on = hp->hd_hostpart;
wp2->wa_spec = wp->wa_spec;
wp->wa_mesg->m_ref++;
wp2->wa_mesg = wp->wa_mesg;
LISTPUTBEFORE(wp, wp2, wa_peer, wa_rpeer);
mp = mesg_new(0);
pkint(mp, wxp->w_ptid);
pkstr(mp, wxp->w_file);
pkint(mp, wxp->w_flags);
pkint(mp, na);
pkint(mp, wxp->w_argc);
for (i = 0; i < wxp->w_argc; i++)
pkstr(mp, wxp->w_argv[i]);
pkint(mp, wxp->w_outtid);
pkint(mp, wxp->w_outctx);
pkint(mp, wxp->w_outtag);
pkint(mp, wxp->w_trctid);
pkint(mp, wxp->w_trcctx);
pkint(mp, wxp->w_trctag);
pkint(mp, wxp->w_nenv);
for (i = 0; i < wxp->w_nenv; i++)
pkstr(mp, wxp->w_env[i]);
pkint(mp, (t+1)-na); /* start proc location 0..count-1 */
pkint(mp, count); /* how many are there in total */
mp->m_dst = hp->hd_hostpart | TIDPVMD;
mp->m_tag = DM_EXEC;
mp->m_wid = wp2->wa_wid;
do {
if (++lasthh > htp->ht_last)
lasthh = 1;
} while (!htp->ht_hosts[lasthh]);
sendmessage(mp);
hp = htp->ht_hosts[lasthh];
na = 0;
}
for (t++ ; t < veclen && vec[t]; t++);
}
return 0;
}
/* tm_sendsig()
*
* Task sending a signal to another task.
*
* TM_SENDSIG
* call {
* int tid
* int signum
* }
* ret { }
*/
int
tm_sendsig(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int tid;
struct pmsg *mp2;
if (upkuint(mp, &tid)) {
pvmlogerror("tm_sendsig() bad msg format\n");
return 0;
}
if (!TIDISTASK(tid)) {
pvmlogprintf("tm_sendsig() bad tid %x\n", tid);
return 0;
}
mp2 = replymessage(mp);
sendmessage(mp2);
mp->m_ref++;
mp->m_src = pvmmytid;
mp->m_dst = (tid & TIDHOST) | TIDPVMD;
mp->m_tag = DM_SENDSIG;
mp->m_wid = 0;
sendmessage(mp);
return 0;
}
/* tm_config()
*
* Task wants machine configuration.
*
* TM_CONFIG
* call { }
* ret {
* int nhosts
* int narches
* {
* int tid
* string name
* string arch
* int speed
* int dsig
* } [nhosts]
* }
*/
int
tm_config(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int hh;
struct hostd *hp;
mp = replymessage(mp);
pkint(mp, hosts->ht_cnt);
pkint(mp, hosts->ht_narch);
for (hh = 1; hh <= hosts->ht_last; hh++) {
if (hp = hosts->ht_hosts[hh]) {
pkint(mp, hp->hd_hostpart);
pkstr(mp, hp->hd_name);
pkstr(mp, hp->hd_arch ? hp->hd_arch : "");
pkint(mp, hp->hd_speed);
pkint(mp, hp->hd_dsig);
}
}
sendmessage(mp);
return 0;
}
/* tm_halt()
*
* Command from task to stop master pvmd.
*
* TM_HALT
* call { }
* ret { }
*/
int
tm_halt(tp, mp)
struct task *tp;
struct pmsg *mp;
{
mp = mesg_new(0);
mp->m_tag = DM_HALT;
mp->m_dst = hosts->ht_hosts[hosts->ht_master]->hd_hostpart | TIDPVMD;
sendmessage(mp);
return 0;
}
/* tm_task()
*
* Task wants a list of tasks.
* If where is a host or task tid, give a list of matching tasks.
* If where is zero, give all running tasks.
*
* TM_TASK
* call {
* int where
* }
* ret {
* int errcode
* {
* int tid
* int ptid
* int hostpart
* int flag
* string a_out
* int pid
* } [count] // implied
* }
*/
int
tm_task(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int where;
struct pmsg *mp2;
struct waitc *wp;
struct waitc *wp2 = 0; /* master waitc of peer group */
int hh;
struct hostd *hp;
if (upkuint(mp, &where)) {
pvmlogerror("tm_task() bad msg format\n");
return 0;
}
mp = replymessage(mp);
mp->m_dst = tp->t_tid;
mp->m_tag = TM_TASK;
if (where) { /* specific host or task requested */
if (!(hp = tidtohost(hosts, where))) {
pkint(mp, PvmNoHost);
sendmessage(mp);
return 0;
}
pkint(mp, 0);
wp = wait_new(WT_TASK);
wp->wa_mesg = mp;
wp->wa_tid = tp->t_tid;
wp->wa_on = hp->hd_hostpart;
mp = mesg_new(0);
mp->m_tag = DM_TASK;
mp->m_dst = hp->hd_hostpart | TIDPVMD;
mp->m_wid = wp->wa_wid;
pkint(mp, where);
sendmessage(mp);
} else { /* all tasks requested */
pkint(mp, 0);
wp2 = wait_new(WT_TASK);
mp->m_ref++;
wp2->wa_mesg = mp;
wp2->wa_tid = tp->t_tid;
mp2 = mesg_new(0);
mp2->m_tag = DM_TASK;
pkint(mp2, 0);
for (hh = hosts->ht_last; hh > 0; hh--) {
if (!hosts->ht_hosts[hh])
continue;
wp = wait_new(WT_TASK);
mp->m_ref++;
wp->wa_mesg = mp;
wp->wa_tid = tp->t_tid;
wp->wa_on = hosts->ht_hosts[hh]->hd_hostpart;
LISTPUTBEFORE(wp2, wp, wa_peer, wa_rpeer);
mp2->m_dst = hosts->ht_hosts[hh]->hd_hostpart | TIDPVMD;
mp2->m_wid = wp->wa_wid;
mp2->m_ref++;
sendmessage(mp2);
}
pmsg_unref(mp2);
pmsg_unref(mp);
/* send message if all waiters are in */
if (wp2->wa_peer == wp2) {
mp->m_ref++;
sendmessage(mp);
}
wait_delete(wp2);
}
return 0;
}
/* tm_tickle()
*
* Task wants to poke at pvmd.
*
* TM_TICKLE
* call {
* int nargs
* int args[nargs]
* }
* ret {
* int nresult
* int results[nresult]
* }
*/
int
tm_tickle(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int nar;
int arg[10];
int i;
char *p;
struct hostd *hp;
upkint(mp, &nar);
if (nar < 1 || nar > 10) {
pvmlogprintf("tm_tickle() bad msg format\n");
return 0;
}
for (i = 0; i < nar; i++)
upkint(mp, &arg[i]);
while (i < sizeof(arg)/sizeof(arg[0]))
arg[i++] = 0;
pvmlogprintf("tm_tickle() #");
for (i = 0; i < nar; i++)
pvmlogprintf(" %d", arg[i]);
pvmlogprintf("\n");
mp = replymessage(mp);
switch (arg[0]) {
case 0:
i_dump(1);
pkint(mp, 0);
break;
case 1:
ht_dump(hosts);
pkint(mp, 0);
break;
case 2:
task_dump();
pkint(mp, 0);
break;
case 3:
wait_dumpall();
pkint(mp, 0);
break;
case 4:
mb_dumpall();
pkint(mp, 0);
break;
case 5:
pkint(mp, 1);
pkint(mp, pvmdebmask);
break;
case 6:
pvmdebmask = arg[1];
pvmlogprintf("tm_tickle() debug mask is %x (%s)\n",
pvmdebmask, debug_flags(pvmdebmask));
pkint(mp, 0);
break;
case 7:
if (arg[1] > 0 && arg[1] < 50) {
nopax = arg[1];
pvmlogprintf("tm_tickle() nopax is %d\n", nopax);
} else
pvmlogprintf("tm_tickle() bogus nopax %d\n", arg[1]);
pkint(mp, 0);
break;
case 8:
pkint(mp, 1);
if ((hp = tidtohost(hosts, arg[1]))
&& hp != hosts->ht_hosts[hosts->ht_local]) {
pvmlogprintf("tm_tickle() failing %s\n", hp->hd_name);
hostfailentry(hp);
ht_delete(hosts, hp);
if (newhosts)
ht_delete(newhosts, hp);
pkint(mp, 1);
} else {
if (hp)
pvmlogprintf("tm_tickle() can't fail %s\n", hp->hd_name);
else
pvmlogprintf("tm_tickle() no such host %x\n", arg[1]);
pkint(mp, 0);
}
break;
case 9:
#ifdef STATISTICS
dump_statistics();
if (arg[1])
reset_statistics();
#else
pvmlogerror("tm_tickle() statistics not compiled in\n");
#endif
pkint(mp, 0);
break;
case 100:
#ifdef SHMEM
peer_dump();
pidtid_dump();
#else
pvmlogerror("tm_tickle() not using shared memory\n");
#endif
pkint(mp, 0);
break;
case 20:
{
struct pmsg *mp2;
int i, j;
for (j = 0; j < arg[1]; j++) {
mp2 = mesg_new(0);
for (i = 0; i < 1000; i++)
pkint(mp2, i);
mp2->m_dst = arg[2];
mp2->m_tag = DM_NULL;
sendmessage(mp2);
}
pkint(mp, 0);
}
break;
default:
pvmlogprintf("tm_tickle() don't know #%d\n", arg[0]);
pkint(mp, 0);
break;
}
sendmessage(mp);
return 0;
}
/* tm_delhost()
*
* Task requesting to add hosts to virtual machine. Exit point is
* here or dm_delhostack().
*
* TM_DELHOST
* call {
* int nhosts
* string names[nhosts]
* }
* ret {
* int nhosts // or negative for error
* int status[nhosts] // status of each host
* }
*/
int
tm_delhost(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int count;
char *buf;
struct waitc *wp;
/* sanity check the message */
if (upkint(mp, &count))
goto bad;
if (count < 1 || count > (tidhmask >> (ffs(tidhmask) - 1)))
goto bad;
while (count-- > 0)
if (upkstralloc(mp, &buf))
goto bad;
else
PVM_FREE(buf);
/* make a wait channel for the task */
wp = wait_new(WT_DELHOST);
wp->wa_tid = tp->t_tid;
wp->wa_on = hosts->ht_hosts[hosts->ht_master]->hd_hostpart;
wp->wa_mesg = replymessage(mp);
/* forward message to master pvmd */
mp->m_ref++;
mp->m_src = pvmmytid;
mp->m_dst = hosts->ht_hosts[hosts->ht_master]->hd_hostpart | TIDPVMD;
mp->m_tag = DM_DELHOST;
mp->m_wid = wp->wa_wid;
sendmessage(mp);
return 0;
bad:
pvmlogprintf("tm_delhost() from t%x bad msg format\n", mp->m_src);
return 0;
}
/* tm_mca()
*
* Task announces multicast address list for subsequent message.
*
* TM_MCA
* call {
* int count // number of addresses
* int tids[count] // addresses
* }
* // No Return
*/
int
tm_mca(tp, mp)
struct task *tp;
struct pmsg *mp;
{
struct mca *mcap; /* mca descriptor */
int ndst; /* num of dst tids */
int *dsts; /* dst tids */
int tid;
int i, j;
/*
* unpack list of dst tids from message (expect it to be sorted).
* discard tids to nonexistent foreign hosts.
*/
mcap = mca_new();
mcap->mc_tid = TIDGID | tp->t_tid;
upkint(mp, &ndst);
dsts = TALLOC(ndst, int, "dsts");
for (i = 0; i < ndst; ) {
upkuint(mp, &tid);
if ((tid & tidhmask) == myhostpart || tidtohost(hosts, tid))
dsts[i++] = tid;
}
ndst = i;
if (ndst < 1)
goto noguys;
/*
* send DM_MCA messages containing tids to destination hosts
* make list of destination hosts for us
*/
mcap->mc_dsts = TALLOC(ndst, int, "mcal"); /* XXX cheat, too much space */
mcap->mc_ndst = 0;
for (j = 0; j < ndst; ) {
i = j;
tid = dsts[i] & tidhmask;
while (++j < ndst && tid == (dsts[j] & tidhmask)) ;
mp = mesg_new(0);
mp->m_dst = (tid |= TIDPVMD);
mp->m_tag = DM_MCA;
pkint(mp, mcap->mc_tid);
pkint(mp, j - i);
while (i < j)
pkint(mp, dsts[i++]);
sendmessage(mp);
mcap->mc_dsts[mcap->mc_ndst++] = tid;
}
noguys:
PVM_FREE(dsts);
/*
* tag task descriptor with mca desc and send mca back to task
*/
tp->t_mca = mcap;
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf("tm_mca() made mca %x for t%x\n",
mcap->mc_tid, tp->t_tid);
}
return 0;
}
/* tm_notify()
*
* Task wants to be notified when an event happens.
*
* TM_NOTIFY
* call {
* int what // event type and flags
* int ctx // message context to use in reply
* int tag // message tag to use in reply
* int count // number of addresses
* int tids[count] // addresses
* }
* // No Return
*/
int
tm_notify(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int what, flags, ctx, tag, count, tid;
struct hostd *hp;
struct pmsg *mp2;
struct waitc *wp;
int i;
if (upkint(mp, &what)
|| upkint(mp, &ctx)
|| upkint(mp, &tag)
|| upkint(mp, &count)) {
pvmlogerror("tm_notify() bad msg format\n");
return 0;
}
flags = what;
what &= 0xff;
switch (what) {
/*
* TaskExit: make a wait context that hangs out until the task is
* cleaned up.
*/
case PvmTaskExit:
for (i = 0; i < count; i++) {
if (upkuint(mp, &tid)) {
pvmlogerror("tm_notify() bad msg format\n");
return 0;
}
if (flags & PvmNotifyCancel) {
FORLIST (wp, waitlist, wa_link)
if (wp->wa_kind == WT_TASKX
&& wp->wa_on == tid
&& wp->wa_tid == tp->t_tid
&& wp->wa_mesg->m_ctx == ctx
&& wp->wa_mesg->m_tag == tag)
break;
if (wp != waitlist) {
sendmessage(wp->wa_mesg);
wp->wa_mesg = 0;
if ((hp = tidtohost(hosts, tid))
&& hp->hd_hostpart != myhostpart)
wp->wa_tid = 0;
else
wait_delete(wp);
}
} else {
/* make reply message */
mp2 = mesg_new(0);
mp2->m_dst = tp->t_tid;
mp2->m_ctx = ctx;
mp2->m_tag = tag;
mp2->m_wid = mp->m_wid;
pkint(mp2, tid);
/* if task doesn't exist, reply immediately */
if (!(hp = tidtohost(hosts, tid))
|| (hp->hd_hostpart == myhostpart && !task_find(tid)))
sendmessage(mp2);
else {
/* otherwise make a wc for it */
wp = wait_new(WT_TASKX);
wp->wa_on = tid;
wp->wa_tid = tp->t_tid;
wp->wa_dep = mp->m_wid;
wp->wa_mesg = mp2;
/* and if not on this host, pass on the request */
if (hp->hd_hostpart != myhostpart) {
mp2 = mesg_new(0);
pkint(mp2, what);
pkint(mp2, tid);
mp2->m_dst = hp->hd_hostpart | TIDPVMD;
mp2->m_tag = DM_NOTIFY;
mp2->m_wid = wp->wa_wid;
sendmessage(mp2);
}
}
}
}
break;
case PvmHostDelete:
for (i = 0; i < count; i++) {
if (upkuint(mp, &tid)) {
pvmlogerror("tm_notify() bad msg format\n");
return 0;
}
if (flags & PvmNotifyCancel) {
if (hp = tidtohost(hosts, tid)) {
FORLIST (wp, waitlist, wa_link)
if (wp->wa_kind == WT_HOSTF
&& wp->wa_on == hp->hd_hostpart
&& wp->wa_tid == tp->t_tid
&& wp->wa_mesg->m_ctx == ctx
&& wp->wa_mesg->m_tag == tag)
break;
if (wp != waitlist) {
sendmessage(wp->wa_mesg);
wp->wa_mesg = 0;
wait_delete(wp);
}
}
} else {
mp2 = mesg_new(0);
mp2->m_dst = mp->m_src;
mp2->m_ctx = ctx;
mp2->m_tag = tag;
pkint(mp2, tid);
if (hp = tidtohost(hosts, tid)) {
wp = wait_new(WT_HOSTF);
wp->wa_tid = tp->t_tid;
wp->wa_on = hp->hd_hostpart;
wp->wa_mesg = mp2;
} else {
sendmessage(mp2);
}
}
}
break;
case PvmHostAdd:
FORLIST (wp, waitlist, wa_link)
if (wp->wa_kind == WT_HOSTA
&& wp->wa_tid == tp->t_tid
&& wp->wa_mesg->m_ctx == ctx
&& wp->wa_mesg->m_tag == tag)
break;
/* if cancelling, delete possible existing waitc */
if (count == 0 || (flags & PvmNotifyCancel)) {
if (wp != waitlist)
wait_delete(wp);
/* otherwise, update existing waitc or create new one */
} else {
if (wp == waitlist) {
wp = wait_new(WT_HOSTA);
wp->wa_tid = tp->t_tid;
wp->wa_on = tp->t_tid;
wp->wa_mesg = mesg_new(0);
wp->wa_mesg->m_dst = tp->t_tid;
wp->wa_mesg->m_ctx = ctx;
wp->wa_mesg->m_tag = tag;
}
wp->wa_count = count;
}
break;
default:
pvmlogprintf("tm_notify() unknown what=%d\n", what);
break;
}
return 0;
}
/* tm_mstat()
*
* Task wants status of a host.
*
* TM_MSTAT
* call {
* string hostname
* }
* ret {
* int status
* }
*/
int
tm_mstat(tp, mp)
struct task *tp;
struct pmsg *mp;
{
char *name;
struct hostd *hp;
struct waitc *wp;
if (upkstralloc(mp, &name)) {
pvmlogerror("tm_mstat() bad msg format\n");
return 0;
}
hp = nametohost(hosts, name);
PVM_FREE(name);
if (!hp) {
mp = replymessage(mp);
pkint(mp, PvmNoHost);
sendmessage(mp);
return 0;
}
wp = wait_new(WT_MSTAT);
wp->wa_tid = tp->t_tid;
wp->wa_on = hp->hd_hostpart;
wp->wa_mesg = replymessage(mp);
mp = mesg_new(0);
mp->m_dst = hp->hd_hostpart | TIDPVMD;
mp->m_tag = DM_PSTAT;
mp->m_wid = wp->wa_wid;
pkint(mp, hp->hd_hostpart | TIDPVMD);
sendmessage(mp);
return 0;
}
/* tm_db()
*
* Task wants a database op.
*
* TM_DB
* call {
* int opcode // insert, delete, lookup
* string name
* int index
* int data // if 'insert'
* }
* ret {
* int index // and status
* int data // if 'lookup'
* }
*/
int
tm_db(tp, mp)
struct task *tp;
struct pmsg *mp;
{
struct waitc *wp;
wp = wait_new(WT_DB);
wp->wa_tid = tp->t_tid;
wp->wa_on = hosts->ht_hosts[hosts->ht_master]->hd_hostpart;
wp->wa_mesg = replymessage(mp);
mp->m_ref++;
mp->m_src = pvmmytid;
mp->m_dst = hosts->ht_hosts[hosts->ht_master]->hd_hostpart | TIDPVMD;
mp->m_tag = DM_DB;
mp->m_wid = wp->wa_wid;
sendmessage(mp);
return 0;
}
/* tm_sched()
*
* Task wants to register as scheduler.
*
* TM_SCHED
* call {
* }
* ret {
* int oldschedtid
* int tid // of master pvmd
* string name
* string arch
* int speed
* int dsig
* }
*/
int
tm_sched(tp, mp)
struct task *tp;
struct pmsg *mp;
{
struct pmsg *mp2;
struct hostd *hp;
mp2 = replymessage(mp);
pkint(mp2, pvmschedtid);
hp = hosts->ht_hosts[hosts->ht_master];
pkint(mp2, hp->hd_hostpart);
pkstr(mp2, hp->hd_name);
pkstr(mp2, hp->hd_arch ? hp->hd_arch : "");
pkint(mp2, hp->hd_speed);
pkint(mp2, hp->hd_dsig);
sendmessage(mp2);
if (pvmdebmask & PDMSCHED) {
pvmlogprintf("tm_sched() old t%x new t%x\n", pvmschedtid, tp->t_tid);
}
pvmschedtid = tp->t_tid;
tp->t_flag |= TF_ISSCHED;
return 0;
}
/* tm_tasker()
*
* Task registers as local task starter.
*
* TM_TASKER
* call {
* int regme // true to register, false to unregister
* }
* ret {
* int code // 0 if okay, else error
* }
*/
int
tm_tasker(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int regme;
struct pmsg *mp2;
if (upkint(mp, ®me)) {
pvmlogerror("tm_tasker() bad msg format\n");
return 0;
}
mp2 = replymessage(mp);
if (regme) {
if (taskertid) {
pkint(mp2, PvmAlready);
} else {
taskertid = tp->t_tid;
tp->t_flag |= TF_ISTASKER;
if (pvmdebmask & PDMTASK) {
pvmlogprintf("tm_tasker() register t%x \n", tp->t_tid);
}
pkint(mp2, 0);
}
} else {
if (taskertid == tp->t_tid) {
taskertid = 0;
tp->t_flag &= ~TF_ISTASKER;
if (pvmdebmask & PDMTASK) {
pvmlogprintf("tm_tasker() unregister t%x \n", tp->t_tid);
}
pkint(mp2, 0);
} else {
if (pvmdebmask & PDMTASK) {
pvmlogprintf("tm_tasker() t%x tries to unregister?\n",
tp->t_tid);
}
pkint(mp2, PvmNoTask);
}
}
sendmessage(mp2);
return 0;
}
/* tm_hoster()
*
* Task registers as slave pvmd starter.
*
* TM_HOSTER
* int regme // true to register, false to unregister
* call {
* }
* ret {
* int code // 0 if okay, else error
* }
*/
int
tm_hoster(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int regme;
struct pmsg *mp2;
if (upkint(mp, ®me)) {
pvmlogerror("tm_hoster() bad msg format\n");
return 0;
}
mp2 = replymessage(mp);
/* always wear 2 condoms, who knows if indices are enough... :) */
if (hosts->ht_hosts[hosts->ht_local]->hd_hostpart
!= hosts->ht_hosts[hosts->ht_master]->hd_hostpart) {
pkint(mp2, PvmHostrNMstr);
sendmessage(mp2);
return 0;
}
if (regme) {
if (hostertid) {
pkint(mp2, PvmAlready);
} else {
hostertid = tp->t_tid;
tp->t_flag |= TF_ISHOSTER;
if (pvmdebmask & PDMSTARTUP) {
pvmlogprintf("tm_hoster() register t%x \n", tp->t_tid);
}
pkint(mp2, 0);
}
} else {
if (hostertid == tp->t_tid) {
hostertid = 0;
tp->t_flag &= ~TF_ISHOSTER;
if (pvmdebmask & PDMSTARTUP) {
pvmlogprintf("tm_hoster() unregister t%x \n", tp->t_tid);
}
pkint(mp2, 0);
} else {
if (pvmdebmask & PDMSTARTUP) {
pvmlogprintf("tm_hoster() t%x tries to unregister?\n",
tp->t_tid);
}
pkint(mp2, PvmNoTask);
}
}
sendmessage(mp2);
return 0;
}
/* tm_tracer()
*
* Task registers as task trace collector.
*
* TM_TRACER
* call {
* int regme // true to register, false to unregister
* int tctx // if regme true, message ctxt for trace events
* int ttag // if regme true, message tag for trace events
* int octx // if regme true, message ctxt for output events
* int otag // if regme true, message tag for output events
* Pvmtmask tmask // if regme true, trace mask for tasks
* int tbuf // if regme true, trace buffer size for tasks
* int topt // if regme true, trace options for tasks
* }
* ret {
* int code // 0 if okay, else error
* }
*/
int
tm_tracer(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int regme;
int tctx;
int ttag;
int octx;
int otag;
Pvmtmask tmask;
int tbuf;
int topt;
struct pmsg *mp2;
struct pmsg *mpd;
int slconf;
int hh;
struct hostd *hp;
char buf[512];
if (upkint(mp, ®me)) {
pvmlogerror("tm_tracer() bad msg format\n");
return 0;
}
mp2 = replymessage(mp);
slconf = 0;
if (regme) {
if (pvmtracer.trctid) {
pkint(mp2, PvmAlready);
} else {
tp->t_flag |= TF_ISTRACER;
upkint(mp, &tctx);
upkint(mp, &ttag);
upkint(mp, &octx);
upkint(mp, &otag);
upkstr(mp, tmask, TEV_MASK_LENGTH);
upkint(mp, &tbuf);
upkint(mp, &topt);
pvmtracer.trctid = tp->t_tid;
pvmtracer.trcctx = tctx;
pvmtracer.trctag = ttag;
pvmtracer.outtid = tp->t_tid;
pvmtracer.outctx = octx;
pvmtracer.outtag = otag;
BCOPY(tmask,pvmtracer.tmask,TEV_MASK_LENGTH);
pvmtracer.trcbuf = tbuf;
pvmtracer.trcopt = topt;
slconf++;
if (pvmdebmask & PDMTRACE) {
pvmlogprintf("tm_tracer() register t%x \n", tp->t_tid);
}
pkint(mp2, 0);
}
} else {
if (pvmtracer.trctid == tp->t_tid) {
tp->t_flag &= ~TF_ISTRACER;
pvmtracer.trctid = 0;
pvmtracer.trcctx = 0;
pvmtracer.trctag = 0;
pvmtracer.outtid = 0;
pvmtracer.outctx = 0;
pvmtracer.outtag = 0;
TEV_MASK_INIT(pvmtracer.tmask);
pvmtracer.trcbuf = 0;
pvmtracer.trcopt = 0;
slconf++;
if (pvmdebmask & PDMTRACE) {
pvmlogprintf("tm_tracer() unregister t%x \n",
tp->t_tid);
}
pkint(mp2, 0);
} else {
if (pvmdebmask & PDMTRACE) {
pvmlogprintf("tm_tracer() t%x tries to unregister?\n",
tp->t_tid);
}
pkint(mp2, PvmNoTask);
}
}
/* Update slave pvmds */
if ( slconf ) {
for (hh = hosts->ht_last; hh > 0; hh--) {
if (hp = hosts->ht_hosts[hh]) {
mpd = mesg_new(0);
mpd->m_tag = DM_SLCONF;
mpd->m_dst = hp->hd_hostpart | TIDPVMD;
pkint(mpd, DM_SLCONF_TRACE);
sprintf(buf, "%x %d %d %x %d %d %d %d %s",
pvmtracer.trctid, pvmtracer.trcctx,
pvmtracer.trctag,
pvmtracer.outtid, pvmtracer.outctx,
pvmtracer.outtag,
pvmtracer.trcbuf, pvmtracer.trcopt,
pvmtracer.tmask);
pkstr(mpd, buf);
sendmessage(mpd);
}
}
}
/* Send response to requesting task */
sendmessage(mp2);
return 0;
}
/* tm_hostsync()
*
* Task wants clock sample from host.
*
* TM_HOSTSYNC
* call {
* int where // pvmd tid
* }
* ret {
* int status // 0 or error code
* int sec // time of day clock
* int usec
* }
*/
int
tm_hostsync(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int where;
struct hostd *hp;
struct pmsg *mp2;
struct waitc *wp;
if (upkuint(mp, &where)) {
pvmlogerror("tm_hostsync() bad msg format\n");
return 0;
}
mp2 = replymessage(mp);
if (!(hp = tidtohost(hosts, where))) {
pkint(mp2, PvmNoHost);
sendmessage(mp2);
return 0;
}
wp = wait_new(WT_HOSTSYNC);
wp->wa_tid = mp->m_src;
wp->wa_on = hp->hd_hostpart;
wp->wa_mesg = mp2;
mp2 = mesg_new(0);
mp2->m_dst = where | TIDPVMD;
mp2->m_tag = DM_HOSTSYNC;
mp2->m_wid = wp->wa_wid;
sendmessage(mp2);
return 0;
}
/* tm_setopt()
*
* Set task parameters
*
* TM_SETOPT
* call {
* {
* int what // which option
* string value // option value
* } [] // implied
* }
* ret {
*/
int
tm_setopt(tp, mp)
struct task *tp;
struct pmsg *mp;
{
int what;
char *val;
int x;
while (!upkint(mp, &what) && !upkstralloc(mp, &val)) {
switch (what) {
case TS_OUTTID:
x = pvmxtoi(val);
change_output(tp, x, tp->t_outctx, tp->t_outtag);
break;
case TS_OUTCTX:
x = pvmxtoi(val);
change_output(tp, tp->t_outtid, x, tp->t_outtag);
break;
case TS_OUTTAG:
x = pvmxtoi(val);
change_output(tp, tp->t_outtid, tp->t_outctx, x);
break;
case TS_TRCTID:
x = pvmxtoi(val);
change_trace(tp, x, tp->t_trcctx, tp->t_trctag);
break;
case TS_TRCCTX:
x = pvmxtoi(val);
change_trace(tp, tp->t_trctid, x, tp->t_trctag);
break;
case TS_TRCTAG:
x = pvmxtoi(val);
change_trace(tp, tp->t_trctid, tp->t_trcctx, x);
break;
default:
pvmlogprintf("tm_setopt() ? option %d val <%s>\n", what, val);
break;
}
if (val)
PVM_FREE(val);
}
mp = replymessage(mp);
sendmessage(mp);
return 0;
}
/* change_output()
*
* Task sets its output sink parameters.
* If new values are different we have to send close and open
* messages, because the collector won't know what's going on.
*/
int
change_output(tp, outtid, outctx, outtag)
struct task *tp;
int outtid, outctx, outtag;
{
struct pmsg *mp;
if (tp->t_outtid != outtid
|| tp->t_outctx != outctx
|| tp->t_outtag != outtag) {
if (tp->t_outtid > 0) {
mp = mesg_new(0);
mp->m_dst = tp->t_outtid;
mp->m_ctx = tp->t_outctx;
mp->m_tag = tp->t_outtag;
pkint(mp, tp->t_tid);
pkint(mp, 0);
sendmessage(mp);
}
if (pvmdebmask & PDMTASK) {
pvmlogprintf("t%x changes output from <t%x %d %d> to <t%x %d %d>\n",
tp->t_tid,
tp->t_outtid, tp->t_outctx, tp->t_outtag,
outtid, outctx, outtag);
}
tp->t_outtid = outtid;
tp->t_outctx = outctx;
tp->t_outtag = outtag;
if (tp->t_outtid > 0) {
mp = mesg_new(0);
mp->m_dst = tp->t_outtid;
mp->m_ctx = tp->t_outctx;
mp->m_tag = tp->t_outtag;
pkint(mp, tp->t_tid);
pkint(mp, -1);
pkint(mp, tp->t_ptid);
sendmessage(mp);
mp = mesg_new(0);
mp->m_dst = tp->t_outtid;
mp->m_ctx = tp->t_outctx;
mp->m_tag = tp->t_outtag;
pkint(mp, tp->t_tid);
pkint(mp, -2);
pkint(mp, tp->t_ptid);
sendmessage(mp);
}
}
return 0;
}
/* tm_getopt()
*
* Connect message for shared memory apps to get trace & output parameters.
* XXX hack, will be merged into something else in v3.4.
*
* TM_GETOPT
* call { }
* ret {
* int tid // task tid
* int ptid // parent tid
* int outtid // output dst
* int outctx
* int outtag
* int trctid // trace dst
* int trcctx
* int trctag
* int udpmax
* int nativecode
* string inaddr
* int schedtid // scheduler tid
* }
*/
int
tm_getopt(tp, mp)
struct task *tp;
struct pmsg *mp;
{
mp = replymessage(mp);
pkint(mp, tp->t_tid);
pkint(mp, tp->t_ptid);
pkint(mp, tp->t_outtid);
pkint(mp, tp->t_outctx);
pkint(mp, tp->t_outtag);
pkint(mp, tp->t_trctid);
pkint(mp, tp->t_trcctx);
pkint(mp, tp->t_trctag);
pkint(mp, pvmudpmtu);
pkint(mp, pvmmydsig);
pkstr(mp, inadport_hex(&(hosts->ht_hosts[hosts->ht_local]->hd_sad)));
pkint(mp, pvmschedtid);
mp->m_flag |= MM_PRIO; /* set priority handling for this message */
sendmessage(mp);
#if defined(SHMEM)
/* now change connection to TF_SHMCONN and try send packets that have
gathered in this tasks sendq */
tp->t_flag &= ~TF_PRESHMCONN;
tp->t_flag |= TF_SHMCONN;
shm_wrt_pkts(tp); /* write any waiting packets */
#endif
return 0;
}
/* change_output()
*
* Task sets its trace sink parameters.
* If new values are different we have to send close and open
* messages, because the collector won't know what's going on.
*/
int
change_trace(tp, trctid, trcctx, trctag)
struct task *tp;
int trctid, trcctx, trctag;
{
struct pmsg *mp;
if (tp->t_trctid != trctid
|| tp->t_trcctx != trcctx
|| tp->t_trctag != trctag) {
if (tp->t_trctid > 0) {
tev_send_endtask(
tp->t_trctid, tp->t_trcctx, tp->t_trctag,
tp->t_tid, tp->t_status,
tp->t_utime.tv_sec, tp->t_utime.tv_usec,
tp->t_stime.tv_sec, tp->t_stime.tv_usec );
}
if (pvmdebmask & PDMTASK) {
pvmlogprintf( "t%x changes trace from <t%x %d %d> to <t%x %d %d>\n",
tp->t_tid,
tp->t_trctid, tp->t_trcctx, tp->t_trctag,
trctid, trcctx, trctag);
}
tp->t_trctid = trctid;
tp->t_trcctx = trcctx;
tp->t_trctag = trctag;
if (tp->t_trctid > 0) {
tev_send_spntask(
tp->t_trctid, tp->t_trcctx, tp->t_trctag,
tp->t_tid, tp->t_ptid );
tev_send_newtask(
tp->t_trctid, tp->t_trcctx, tp->t_trctag,
tp->t_tid, tp->t_ptid, 0 /* XXX lie */,
tp->t_a_out ? tp->t_a_out : "" );
}
}
return 0;
}
/* tm_context()
*
* Assign task a communication context, or free one in use
*
* TM_CONTEXT
* call {
* int cid // to free, or zero if requesting one
* }
* ret {
* int cid // new cid or status
* }
*/
int
tm_context(tp, mp)
struct task *tp;
struct pmsg *mp;
{
struct pmsg *mp2;
struct ccon *cp;
int cid;
int cc;
mp2 = replymessage(mp);
if (upkuint(mp, &cid)) {
cc = PvmBadParam;
} else {
if (cid == 0) {
if (cp = ccon_new(tp))
cc = cp->c_cid;
else
cc = PvmOutOfRes;
} else {
cc = PvmNotFound;
if (tp->t_ccs) {
FORLIST (cp, tp->t_ccs, c_peer) {
if (cp->c_cid == cid) {
LISTDELETE(cp, c_link, c_rlink);
LISTDELETE(cp, c_peer, c_rpeer);
PVM_FREE(cp);
cc = 0;
break;
}
}
}
}
}
pkint(mp2, cc);
sendmessage(mp2);
return 0;
}
/* tm_shmconn()
*
* Task connecting to pvmd under the shmem system.
* This is used to tie the task to the pvmd socket into the correct
* task struct. This socket is a one-way only socket that is used
* by the task to kick the daemon when the task has put a message
* for it in its shared memory inbox segment. The socket is also
* used to catch when the task dies....
*
* There is no d-auth file checking.
* Only data passed is the task's pid so we can match it with our
* own table.
*
* TM_SHMCONN()
* call {
* int unixpid // real pid
* }
* ret {
* }
*/
int
tm_shmconn(tp, mp)
struct task *tp;
struct pmsg *mp; /* remember no return message */
{
int tid;
int pid; /* real pid of task */
int ppid; /* probable pid */
struct task *tp2; /* to search for real context */
struct pmsg *mp2;
int cc;
char c;
if ( upkint( mp, &ppid ) ) {
pvmlogerror(
"tm_shmconn() bad msg format. Missing Unix pid??\n" );
goto bail;
}
#ifndef SHMEM
if ( 1 ) {
pvmlogerror( "tm_shmconn() shmem conn on non shmem sys? " );
pvmlogerror( "Bit bucket...\n" );
goto bail;
}
#endif
/*
* Task should already have a correct task struct.
* Because TM_SHMCONN is passed after the shmem connect phase
* in pvmbeatask() in lpvmshmem.c
*/
if ( tp2 = task_findpid( ppid ) )
{
if ( pvmdebmask & PDMTASK ) {
pvmlogprintf(
"tm_shmconn() reconnect task t%x via sockets\n",
tp2->t_tid);
}
/* tp->t_sched = tp2->t_sched; */
}
else
{
pvmlogerror( "tm_shmconn() conn from non shmem connd task?\n");
goto bail;
}
/*
* tp2 is now the real task....
* So we now update it with the correct socket descripter and then
* mark the shadow one as dead.
*
*/
tp2->t_sock = tp->t_sock;
tp->t_sock = -1; /* tp will be freed by loclinput() */
tp->t_rxp = 0;
tp->t_flag |= TF_CLOSE; /* stop work() complaining - dead socket? */
/*
* kick it in the butt; it's ready to go
*/
/* We are happy to mark it as a shared memory connected process */
/* i.e. please use shmem (mpp_output) not my socket to talk to me */
/* the PRESHMCONN flag indicates that messages with MM_PRIO can be
sent over shared memory, but no others. tm_getopt will
set the flag to SHMCONN
*/
tp2->t_flag |= TF_PRESHMCONN;
return 0;
bail:
/* i.e. kill it! */
tp->t_sock = -1; /* tp will be freed by loclinput() */
tp->t_rxp = 0;
tp->t_flag |= TF_CLOSE;
return 0;
}